package com.example

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

object DataStreamDemo1 {

  //  def main(args: Array[String]) {
  //    val env = StreamExecutionEnvironment.getExecutionEnvironment
  //    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  //    val input=env.fromCollection(List(("a",1L),("b",1L),("b",5L),("b",5L)))
  //    val timeWindow=input.assignAscendingTimestamps(t=>t._2)
  //    val result=timeWindow.keyBy(0).timeWindow(Time.milliseconds(4)).sum("_2")
  //    result.print()
  //    env.execute()
  //  }

  //  nc -lk 9999
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val socketDs: DataStream[String] = senv.socketTextStream("192.168.88.136", 9999)
    val keyedsteam: DataStream[(String, Int)] = socketDs.flatMap(_.split(" ")).map((_, 1))
    val keySteamres: KeyedStream[(String, Int), String] = keyedsteam.assignAscendingTimestamps(t => t._2).keyBy(value => value._1)
    val res: WindowedStream[(String, Int), String, TimeWindow] = keySteamres.window(TumblingEventTimeWindows.of(Time.seconds(5)))
    val result: DataStream[(String, Int)] = res.sum(1)
    result.print()
    senv.execute()

  }

}
